package rpc.remote.transport.netty.code;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import rpc.enums.CompressTypeEnum;
import rpc.enums.SerializationTypeEnum;
import rpc.extension.ExtensionLoader;
import rpc.compress.Compress;
import rpc.remote.constants.RpcConstants;
import rpc.remote.dto.RpcMessage;
import rpc.remote.dto.RpcRequest;
import rpc.remote.dto.RpcResponse;
import rpc.serialize.Serializer;

import java.util.Arrays;

/**
 * @ClassName: RpcMessageDecoder
 * custom protocol decoder
 * <pre>
 *   0     1     2     3     4        5     6     7     8         9          10      11     12  13  14   15 16
 *   +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+
 *   |   magic   code        |version | full length         | messageType| codec|compress|    RequestId       |
 *   +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+
 *   |                                                                                                       |
 *   |                                         body                                                          |
 *   |                                                                                                       |
 *   |                                        ... ...                                                        |
 *   +-------------------------------------------------------------------------------------------------------+
 * 4B  magic code（魔法数）   1B version（版本）   4B full length（消息长度）    1B messageType（消息类型）
 * 1B compress（压缩类型） 1B codec（序列化类型）    4B  requestId（请求的Id）
 * body（object类型数据）
 * </pre>
 * <p>
 * {@link LengthFieldBasedFrameDecoder} is a length-based decoder , used to solve TCP unpacking and sticking problems.
 * </p>
 * @see <a href="https://zhuanlan.zhihu.com/p/95621344">LengthFieldBasedFrameDecoder解码器</a>
 * @Description:
 * @author: Kowk
 * @since:1.0.0
 */
@Slf4j
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {

    public RpcMessageDecoder(){
        //长度字段偏移：魔法数4B，版本1B然后就是总长度，所以这里一共5B
        //长度字段的长度：总长度full length占4B，所以这里4B
        //长度调整：总长度包含了所有数据，在这之前已经读取了9字节，所有剩下的长度是full length - 9，所以这里的值是-9
        //需要剥离掉的初始字节:由于我们使用的是手动检查魔法数和版本代码，所里这里不需要使用
        this(RpcConstants.MAX_FRAME_LENGTH,5,4,-9,0);
    }
    /**
     * @Description:
     *
     *      * @param maxFrameLength      Maximum frame length. It decide the maximum length of data that can be received.
     *      *                            If it exceeds （超出）, the data will be discarded（丢弃）.
     *      * @param lengthFieldOffset （跳过指定长度字节）   Length field offset. The length field is the one that skips the specified length of byte.
     *      * @param lengthFieldLength   The number of bytes in the length field.
     *      * @param lengthAdjustment    The compensation value（补偿值） to add to the value of the length field 一般要使总长度符合4的倍数
     *      * @param initialBytesToStrip Number of bytes skipped.
     *      *                            If you need to receive all of the header+body data, this value is 0
     *      *                            if you only want to receive the body data, then you need to skip the number of bytes consumed by the header.
     *
     * @return:
     */
    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,  int lengthAdjustment, int initialBytesToStrip ){
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);

    }



    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx,in);
        if (decoded instanceof ByteBuf){
            ByteBuf frame = (ByteBuf) decoded;
            if (frame.readableBytes() >= RpcConstants.TOTAL_LENGTH) {
                try {
                    return decodeFrame(frame);
                }catch (Exception e){
                    log.error("解码框架错误");
                    throw e;
                } finally {
                    frame.release();
                }
            }
        }
        return decoded;
    }

    private Object decodeFrame(ByteBuf in ) {
        //必须要按顺序读
        checkMagicNumber(in);
        checkVersion(in);
        int fullLength = in.readInt();
        // 构造 RpcMessage 对象
        byte messageType = in.readByte();
        byte codecType = in.readByte();
        byte compressType = in.readByte();
        int requestId = in.readInt();
        RpcMessage rpcMessage = RpcMessage.builder()
                .codec(codecType)
                .requestId(requestId)
                .messageType(messageType).build();
        if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
            rpcMessage.setData(RpcConstants.PING);
            return rpcMessage;
        }
        if (messageType == RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
            rpcMessage.setData(RpcConstants.PONG);
            return rpcMessage;
        }
        int bodyLength = fullLength - RpcConstants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bs = new byte[bodyLength];
            in.readBytes(bs);
            // 解压字节流
            String compressName = CompressTypeEnum.getName(compressType);
            Compress compress = (Compress) ExtensionLoader.getExtensionLoader(Compress.class)
                    .getExtension(compressName);
            bs = compress.decompress(bs);
            // 将对象反序列化
            String codecName = SerializationTypeEnum.getName(rpcMessage.getCodec());
            log.info("codec name: [{}] ", codecName);
            Serializer serializer = (Serializer) ExtensionLoader.getExtensionLoader(Serializer.class)
                    .getExtension(codecName);
            if (messageType == RpcConstants.REQUEST_TYPE) {
                RpcRequest tmpValue = serializer.deserialize(bs, RpcRequest.class);
                rpcMessage.setData(tmpValue);
            } else {
                RpcResponse tmpValue = serializer.deserialize(bs, RpcResponse.class);
                rpcMessage.setData(tmpValue);
            }
        }
        return rpcMessage;
    }


    private void checkMagicNumber(ByteBuf in) {
        //读取版本信息并处理
        byte version = in.readByte();
        if (version != RpcConstants.VERSION) {
            throw new RuntimeException("版本不匹配"+version);
        }
    }

    private void checkVersion(ByteBuf in) {
        // 前四位就是魔法数，读取并比较
        int len = RpcConstants.MAGIC_NUMBER.length;
        byte[] tmp = new byte[len];
        in.readBytes(tmp);
        for (int i = 0; i < len; i++) {
            if (tmp[i] != RpcConstants.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("Unknown magic code: " + Arrays.toString(tmp));
            }
        }
    }
}
